
/*
 * Copyright © 2021 https://www.cestc.cn/ All rights reserved.
 */

package com.zx.learn.flink.sink;

import org.apache.flink.api.common.serialization.SimpleStringEncoder;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.core.fs.Path;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig;
import org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink;
import org.apache.flink.streaming.api.functions.sink.filesystem.bucketassigners.DateTimeBucketAssigner;
import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy;

public class StreamingFileSinkDemo {
    public static void main(String[] args) throws Exception {
        //1．初始化流计算运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //2．设置Checkpoint（10s）周期性启动 和 stateBackend 存储路径
        // Sink保证仅一次语义使用 checkpoint 和 二段提交
        env.enableCheckpointing(10000);
        env.setStateBackend(new FsStateBackend("file:///d:/etl/"));

        //4．接入socket数据源，获取数据
        DataStreamSource<String> source = env.socketTextStream("localhost", 9999);
        //5．创建Streamingfilesink对象
        OutputFileConfig config = OutputFileConfig
                .builder()
                .withPartPrefix("crm")
                .withPartSuffix(".txt")
                .build();
        //5-1. 创建输出文件配置，指定输出路径 /FlinkStreamFileSink/parquet
        StreamingFileSink sink = StreamingFileSink.forRowFormat(new Path("hdfs://node1:8020/FlinkStreamFileSink/parquet"),
                new SimpleStringEncoder<String>("UTF-8"))
                // sink-kafka new FlinkKafkaProducer
                //5-2.StreamingFileSink 行格式化 ， withBucketAssigner->DateTimeBucketAssigner
                .withBucketAssigner(new DateTimeBucketAssigner())
                //withRollingPolicy -> 默认滚筒策略
                .withRollingPolicy(DefaultRollingPolicy.builder()
                        .withMaxPartSize(128 * 1024 * 1024)
                        .withRolloverInterval(Time.seconds(2).toMilliseconds())
                        .withInactivityInterval(Time.seconds(2).toMilliseconds())
                        .build())
                //withOutputFileConfig -> 输出文件的配置
                .withOutputFileConfig(config)
                .build();
        //6.设置输出 sink
        source.print();
        source.addSink(sink).setParallelism(1);
        //7．执行任务
        env.execute();
    }
}
